package pro.coman404.tools;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.source.TransactionSource;

/**
 * @author NotFound404 at 2024/7/1
 * code code always code
 */
public class FraudDetectionJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        TransactionSource transactionSource = new TransactionSource();
        DataStream<Transaction> transaction =  env.addSource(transactionSource).name("transaction-source");

        DataStream<Alert> alert = transaction.keyBy(Transaction::getAccountId)
                .process(new FraudDetector())
                .name("fraud-detector");

        alert.addSink(new AlertSink()).name("alert-sender");

        env.execute("Fraud Dectection");
    }

}
